/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 * MMAP queue
 *
 * We store the array queue in MMAP
 * If we use 64bit-version system, we could get more than 2GB MMAP space
 * Fixed-length support only now...
 *
 * ADDITIONAL FILE OF id5db
 * By Dr.NP <np@bsgroup.org>
 */

#include "memcached.h"
#include "id5db_queue.h"
#include "id5db_script.h"
#include "id5db_misc.h"

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <fcntl.h>
#include <dirent.h>
#include <pthread.h>

//struct queue_block_t queue_block[MAX_MMAP_FILES];
char *data_dir = NULL;
size_t item_size = 0;
volatile uint64_t head_id = 0;
volatile uint64_t tail_id = 0;
struct queue_block_t head_block;
struct queue_block_t tail_block;
pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER;

static int selector_one(const struct dirent *entry)
{
    unsigned long int t_id = strtoul(entry->d_name, NULL, 16);
    static char t_filename[_POSIX_PATH_MAX];
    
    if (0 == strcmp(entry->d_name, ".") || 0 == strcmp(entry->d_name, ".."))
    {
        return 0;
    }
    
    snprintf(t_filename, _POSIX_PATH_MAX - 1, "%08lX%s", t_id, MMAP_FILE_SUFFIX);
    
    if (0 == strcmp(t_filename, entry->d_name))
    {
        return 1;
    }
    
    return 0;
}

inline static int get_item_len(char *data)
{
    static int val;
    memcpy(&val, data, sizeof(int));
    
    return val;
}

static uint64_t determine_item_id()
{
    uint64_t val = 0;
    int tail_block_id = -1;
    int head_block_id = -1;
    tail_block.block_id = -1;
    head_block.block_id = -1;
    tail_block.mmap_fd = -1;
    head_block.mmap_fd = -1;
    
    if (!data_dir)
    {
        return -1;
    }
    
    struct dirent **eps;
    int n;
    
    n = scandir(data_dir, &eps, selector_one, alphasort);
    if (n >= 0)
    {
        int i;
        for (i = 0; i < n; i ++)
        {
            if (tail_block_id < 0)
            {
                tail_block_id = atoi(eps[i]->d_name);
            }
        }
        
        if (n > 0)
        {
            head_block_id = atoi(eps[n - 1]->d_name);
        }
    }
    
    else
    {
        fprintf(stderr, "Cannot read content from directory %s\n", data_dir);
        exit(1);
    }
    
    tail_id = get_last_insert_id();
    if (tail_id == 0xFFFFFFFFFFFFFFFF)
    {
        // Query failed
        if (tail_block_id > 0)
        {
            tail_id = tail_block_id * (MMAP_FILESIZE / item_size);
        }
        
        else
        {
            tail_block_id = 0;
            tail_id = 0;
        }
        
        if (head_block_id <= tail_block_id)
        {
            head_block_id = tail_block_id;
            head_id = tail_id;
        }
        
        else
        {
            head_id = (uint64_t) head_block_id * (uint64_t) (MMAP_FILESIZE / item_size);
        }
    }
    
    else
    {
        tail_block_id = (uint64_t) tail_id / (uint64_t) (MMAP_FILESIZE / item_size);
        if (head_block_id <= tail_block_id)
        {
            head_block_id = tail_block_id;
        }
        
        head_id = (uint64_t) (head_block_id + 1) * (uint64_t) (MMAP_FILESIZE / item_size);
    }
    
    id5db_log("Queue - Determine head_id : %llu, tail_id : %llu", head_id, tail_id);
    
    check_mmap_by_item(tail_id, &tail_block);
    check_mmap_by_item(head_id, &head_block);
    
    return val;
}

void id5db_queue_init(const char *ddir, size_t isize)
{
    DIR *dh = opendir(ddir);
    if (!dh)
    {
        // Create a new directory
        if (0 != mkdir(ddir, S_IRWXU))
        {
            fprintf(stderr, "Cannot create directory %s\n", ddir);
            exit(1);
        }
    }
    
    else
    {
        // Directory readable
        closedir(dh);
    }
    
    data_dir = (char *) ddir;
    if (isize > MMAP_FILESIZE || isize <= 0)
    {
        item_size = MMAP_FILESIZE;
    }
    
    else
    {
        item_size = isize;
    }
    
    determine_item_id();
    
    // Start first circle
    dispatch_queue_data();
    
    return;
}

char * check_mmap_by_item(uint64_t item_id, struct queue_block_t *block)
{
    if (!item_size)
    {
        return NULL;
    }
    
    int block_id = (item_id / (MMAP_FILESIZE / item_size));
    struct stat st;
    static char filename[_POSIX_PATH_MAX];
    char *addr = NULL;
    int fd = -1;
    
    if (block_id == tail_block.block_id)
    {
        addr = tail_block.mmap_addr;
        if (!block->mmap_addr || block->mmap_addr != tail_block.mmap_addr)
        {
            block->mmap_addr = addr;
            block->mmap_fd = tail_block.mmap_fd;
            strncpy(block->mmap_filename, tail_block.mmap_filename, _POSIX_PATH_MAX);
            strncpy(block->mmap_filename_c, tail_block.mmap_filename_c, _POSIX_PATH_MAX);
        }
    }
    
    else if (block_id == head_block.block_id)
    {
        addr = head_block.mmap_addr;
        if (!block-> mmap_addr || block->mmap_addr != head_block.mmap_addr)
        {
            block->mmap_addr = addr;
            block->mmap_fd = head_block.mmap_fd;
            strncpy(block->mmap_filename, head_block.mmap_filename, _POSIX_PATH_MAX);
            strncpy(block->mmap_filename_c, head_block.mmap_filename_c, _POSIX_PATH_MAX);
        }
    }
    
    else
    {
        // Open MMAP
        snprintf(filename, _POSIX_PATH_MAX - 1, "%s/%08X%s", data_dir, block_id, MMAP_FILE_SUFFIX);
        fd = open(filename, O_RDWR | O_CREAT, (mode_t) 0600);
        if (-1 == fd)
        {
            // Open MMAP error
            id5db_log("Queue - Cannot open/create MMAP file %s", filename);
            
            return NULL;
        }
        
        stat(filename, &st);
        size_t filesize = st.st_size;
        if (filesize < MMAP_FILESIZE)
        {
            int res = lseek(fd, MMAP_FILESIZE - 1, SEEK_SET);
            if (-1 == res)
            {
                // Resize error
                close(fd);
                return NULL;
            }
            
            write(fd, "", 1);
            fsync(fd);
            id5db_log("Queue - New MMAP file %s created", filename);
        }
        
        addr = mmap(0, MMAP_FILESIZE, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
        if (!addr)
        {
            close(fd);
            return NULL;
        }
        
        if (tail_block.block_id != head_block.block_id && block->mmap_fd > 0)
        {
            id5db_log("Queue - Closing file %s", block->mmap_filename);
            close(block->mmap_fd);
            id5db_file_compress(block->mmap_filename, block->mmap_filename_c);
        }
        
        block->block_id = block_id;
        block->mmap_fd = fd;
        block->mmap_addr = addr;
        strncpy(block->mmap_filename, filename, _POSIX_PATH_MAX - 1);
        snprintf(block->mmap_filename_c, _POSIX_PATH_MAX - 1, "%s/%08X%s", data_dir, block_id, MMAP_FILE_COMPRESSED_SUFFIX);
    }
    
    return addr ? (addr + (item_id * item_size % MMAP_FILESIZE)) : NULL;
}

// Add data to queue
int id5db_queue_store_item(char *data, int *len)
{
    pthread_mutex_lock(&queue_lock);
    char *addr = check_mmap_by_item(head_id, &head_block);
    if (!addr)
    {
        return -1;
    }
    
    head_id ++;
    pthread_mutex_unlock(&queue_lock);
    
    if (*len > item_size - 4)
    {
        *len = item_size - 4;
    }
    
    memcpy(addr, (char *) len, 4);
    memcpy(addr + 4, data, *len);
    
    return (head_id - tail_id);
}

// Fetch data from queue
char * id5db_queue_get_item(int *len, uint64_t *id)
{
    if (tail_id >= head_id)
    {
        *len = 0;
        return NULL;
    }
    
    pthread_mutex_lock(&queue_lock);
    char *addr = check_mmap_by_item(tail_id, &tail_block);
    *len = get_item_len(addr);
    *id = tail_id;
    pthread_mutex_unlock(&queue_lock);
    
    tail_id ++;
    //fprintf(stderr, "%llu : %p --\n", (unsigned long long) tail_id, addr);
    return addr + 4;
}
